package com.corn.flink.lesson4;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * FlinkWatermarkCustomPeriodicStrategyDemo
 *
 * @author JimWu
 * @date 2023/3/1 18:01
 **/
public class FlinkWatermarkCustomPeriodicStrategyDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Event> ds = env.fromElements(
                new Event("add", 1000),
                new Event("delete", 2000),
                new Event("update", 3000),
                new Event("select", 5000),
                new Event("add", 6000),
                new Event("delete", 7000),
                new Event("select", 8000)
        );
        env.getConfig().setAutoWatermarkInterval(500);

        // 自定义水位线生成策略
        ds.assignTimestampsAndWatermarks(new WatermarkStrategy<Event>() {
            @Override
            public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new WatermarkGenerator<Event>() {
                    private long delayTime = 5000;

                    private long maxTimestamp = 0;

                    // 当接受到数据的时候回调
                    @Override
                    public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {
                        maxTimestamp = Math.max(event.timestamp, maxTimestamp);
                    }

                    // 周期性的输出一个水位线 默认200ms 可以通过env.getConfig().setAutoWatermarkInterval()方法设置
                    // 最大数据时间戳 延迟5s的数据
                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {
                        output.emitWatermark(new Watermark(maxTimestamp - delayTime - 1));
                    }
                };
            }
        });

        env.execute();
    }
}
